package com.spark.streaming

import org.apache.spark.{HashPartitioner, SparkConf}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * kafka 
 */
object KafkaTest2 {
  
  /**
    * 第一个参数：聚合的key，就是单词
    * 第二个参数：当前批次产生批次该单词在每一个分区出现的次数
    * 第三个参数：初始值或累加的中间结果
    */
  val updateFunc = (iter: Iterator[(String, Seq[Int], Option[Int])]) => {
    //iter.map(t => (t._1, t._2.sum + t._3.getOrElse(0)))
    iter.map{ case(x, y, z) => (x, y.sum + z.getOrElse(0))}
  }
  
  def main(args: Array[String]): Unit = {
    
    val conf = new SparkConf().setMaster("local[*]").setAppName("KafkaTest1")
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(5))
    
    //如果要使用课更新历史数据（累加），那么就要把终结结果保存起来
    ssc.checkpoint("./ck")
    
    val zk = "localhost:2181"
    val groupId = "g1" // test-consumer-group
    val topics = Map[String,Int]("test" -> 1)
    val data : ReceiverInputDStream[(String, String)] =  KafkaUtils.createStream(ssc, zk, groupId, topics);
    // 取出value值
    val lines : DStream[String] = data.map(_._2)
    // 切分压平
    val words : DStream[String] = lines.flatMap(_.split(" "));
    // 组合元组
    val wordAndOne : DStream[(String,Int)] = words.map((_,1))
    
    //聚合
    val reduced: DStream[(String, Int)] = wordAndOne.updateStateByKey(updateFunc, new HashPartitioner(ssc.sparkContext.defaultParallelism), true)
    //打印结果(Action)
    reduced.print()
    
    ssc.start()
    ssc.awaitTermination()
    
    
  }
}